package com.practice.car.cardataapp.schelduer.spark;

import com.practice.car.cardataapp.schelduer.spark.scheduler.ComplaintsScheduler;
import com.practice.car.cardataapp.schelduer.spark.scheduler.RecomScheduler;
import com.practice.car.cardataapp.schelduer.spark.scheduler.SaleScheduler;
import com.practice.car.cardataapp.schelduer.spark.task.complaints.*;
import com.practice.car.cardataapp.schelduer.spark.task.complaints.ComplaintsRddApplicaiton;
import com.practice.car.cardataapp.schelduer.spark.task.recom.CarGradeAPP;
import com.practice.car.cardataapp.schelduer.spark.task.sale.*;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.*;

@Component
@Data
public class Schedule {
    public static String driver;
    public static final Object lock = new Object();


    @Value("${spring.datasource.druid.driver-class-name}")
    public void setDriver(String driver_class_name) {
        driver = driver_class_name;
    }

    public static String url;
    @Value("${spring.datasource.druid.url}")
    public void setUrl(String url1) {
        url = url1;
    }

    public static String user;
    @Value("${spring.datasource.druid.username}")
    public void setUserName(String userName) {
        user = userName;
    }

    public static String password;
    @Value("${spring.datasource.druid.password}")
    public void setPassword(String password1) {
        password = password1;
    }

    public static String sparkMaster;
    @Value("${scheduler.spark.master}")
    public void setSparkMaster(String sparkMaster1) {
        sparkMaster = sparkMaster1;
    }

    public static String saleHdfsUrl;
    @Value("${scheduler.hadoop.saleUrl}")
    public void setSaleHdfsUrl(String saleHdfsUrl1) {
        saleHdfsUrl = saleHdfsUrl1;
    }

    public static String complaintsHdfsUrl;
    @Value("${scheduler.hadoop.complaintsUrl}")
    public void setcomplaintsHdfsUrl(String complaintsHdfsUrl1) {
        complaintsHdfsUrl = complaintsHdfsUrl1;
    }

    // 2020/9/13 0013 增加推荐模块的所需的json
    public static String recomHdfsUrl;
    @Value("${scheduler.hadoop.recomUrl}")
    public void setrecomHdfsUrl(String recomHdfsUrl1) {
        recomHdfsUrl = recomHdfsUrl1;
    }

    public static String subCarUrl;
    @Value("${scheduler.hadoop.subCarUrl}")
    public void setSubcarUrl(String subCarUrl1) {
        subCarUrl = subCarUrl1;
    }

    public static Connection connection = null;

    static Timer timer ;
    private Integer urltest;
    public static  boolean isContinue = false;
    public static boolean isRun = false;

    public Schedule() {
        monitor();
    }

    private void monitor() {
        Timer timer = new Timer();
        long intevalPeriod = 24* 60 * 60 * 1000;//7*60000ms=7min调度一次  一天检查一次是否要停下或者启动
        timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                if (Schedule.isContinue&& Schedule.timer == null) {
                    Schedule.run();
                } else if (!Schedule.isContinue&& Schedule.timer != null) {
                    Schedule.timer.cancel();
                    Schedule.timer=null;
                }
            }
        }, 0L, intevalPeriod);
    }

    public static void begin() {
        isContinue = true;
        if (timer != null) {
            if (Schedule.isRun) {
                return;
            }
            timer.cancel();
            timer.purge();
            timer = null;
        }
        run();
    }
    private static void run() {
        timer = new Timer();
        System.out.println("start time = " + new Date());
        TimerTask task = new TimerTask() {
            @Override
            public void run() {
                Schedule.isRun = true;
                synchronized (Schedule.lock) {
                    SaleScheduler saleScheduler = new SaleScheduler();
                    ComplaintsScheduler complaintsScheduler = new ComplaintsScheduler();
                    RecomScheduler recomScheduler = new RecomScheduler();
                    saleScheduler.add(new CarPrice());
                    saleScheduler.add(new RddApplicaiton());
                    saleScheduler.add(new App());
                    saleScheduler.add(new suvtest());
                    saleScheduler.add(new TestDB());

                    complaintsScheduler.add(new ComplaintsAnalysis());
                    complaintsScheduler.add(new ComplaintsRddApplicaiton());

                    recomScheduler.add(new CarGradeAPP());

                    try {
                        initData();
                        saleScheduler.schedule();
                        complaintsScheduler.schedule();
                        recomScheduler.schedule();

                        if (connection != null) {
                            connection.close();
                        }
                        System.out.println("end time = " + new Date());
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }finally {
                        Schedule.isRun = false;
                    }
                }
            }
        };
        long delay = 0;
        long intevalPeriod = 20 * 24 * 60 * 60 * 1000;//7*60000ms=7min调度一次
        timer.scheduleAtFixedRate(task, delay, intevalPeriod);
    }


    private static void initData() {
        System.out.println("---------init database---------");
        //预处理添加数据:删除表中存在的数据
        List<String> list = Arrays.asList("brand_sales",
                "car_price", "car_sale_month", "car_sale_trend", "mpv_national_sales",
                "national_car_sales_trends", "suv_and_mpv_sales_ratio", "suv_national_sales",
                "total_national_sales","car_sale_month_top10","car_place_complaint","complaint","complaint_ranking","subcar");
        list.forEach(e-> {
            try {
                getConnection().prepareStatement("truncate table "+e).executeUpdate();
            } catch (SQLException ex) {
                ex.printStackTrace();
            }
        });
    }
    public static  Connection getConnection() {
        //获取数据库连接
        try {
            if (connection != null&&!connection.isClosed()) {
                return connection;
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
        String username = Schedule.user;
        String password = Schedule.password;
        String url = Schedule.url;
        String driver = Schedule.driver;
        try {
            connection = DriverManager.getConnection(url, username, password);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return connection;
    }

}
